package day1

import org.apache.log4j.{Level, Logger}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object RDD {
  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "D:\\新建文件夹\\HCIA初级\\hadoop")
    Logger.getLogger("org").setLevel(Level.OFF)
    val sc = new SparkContext(new SparkConf().setAppName("Basic").setMaster("local"))

    /*
     * RDD算子分为：RDD、PairRDD、数值RDD
     * RDD：标准RDD，PairRDD、数值RDD来自标准RDD
     * RDD中的转换算子对于PairRDD、数值RDD都可以使用，PairRDD、数值RDD的转换算子大部分RDD无法使用
     */

    val arr = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
    val rdd1 = sc.parallelize(arr)
    rdd1.persist(StorageLevel.MEMORY_ONLY)

    //RDD中的转换算子
    //map()算子：对于一个RDD做一对一遍历，将前一个RDD的元素读取出来处理完成之后导入到新的RDD中
    //前一个RDD和后一个RDD的长度相同
    val rdd1_map = rdd1.map(_ * 3)

    //union()算子：合并RDD，先将前一个RDD的元素赋值给当前RDD，再将后一个元素赋值给当前RDD
    //当前新创建的RDD长度为两个RDD的长度总和
    val rdd1_union = rdd1.union(rdd1_map)

    //filter()算子：过滤、筛选，当前算子需要传递一个布尔值;当前值为true时将这个数据获取
    //会减少rdd的长度
    val rdd1_filter = rdd1.filter(_ % 2 == 0)

    //flatMap()算子：扁平化操作，参数、函数的返回值都为容器，容器返回的成员为当前RDD的元素
    val rdd1_flatMap = rdd1.flatMap(1 to _)
    val rdd2 = sc.parallelize(List("hello world","hello scala","hello spark"))
    val rdd2_flatMap = rdd2.flatMap(_.split(" "))

    //intersection()：交集，将两个RDD的内容进行处理，将相同的数据导出为新RDD的元素
    val rdd1_intersection = rdd1.intersection(rdd1_filter)
    val rdd3 = sc.parallelize(List("hello world","hello scala"))
    val rdd3_intersection = rdd2.intersection(rdd3)

    //subtract()算子：前一个RDD减去后一个RDD的内容，返回一个不包含后一个RDD元素的前一个RDD元素
    val rdd1_subtract = rdd1.subtract(rdd1_filter)

    //cartesian()算子：笛卡儿积，将两个元素合并，以前一个RDD的每一个元素作为key，后一个RDD的每一个元素作为value，
    // 然后生成一个新的PairRDD
    val rdd1_cartesian = rdd1.cartesian(rdd2_flatMap)

    //mapPartitions()算子：根据要求生成对应RDD，通过迭代器获得对应的数值
    //迭代器特性：读一个丢一个
    rdd1.mapPartitions(iter => {
      val list = iter.toList
      val even = list.filter(_ % 2 == 0).sum
      val odd = list.filter(_ % 2 != 0).sum
      Iterator(even,odd)
    }
    )
    //      .foreach(println)

    //mapPartitionsWithIndex()算子：针对不同的分区做操作，简化RDD的复杂程度
    //val rdd2 = sc.parallelize(List("hello world","hello scala","hello spark"))
    val rdd5 = sc.parallelize(List("hello world","hello scala","hello spark"),2)
    rdd5.mapPartitionsWithIndex((index , iter) => {
      println(s"这是第 $index 分区操作")
      iter.map(println(_))
    }
    )
    //      .count()

    //RDD中的行动算子
    //count()算子：遍历出里面每个数据
    //foreach()算子：定义一个输出方式，将对于RDD中的内容进行遍历输出
    //    rdd1_map.foreach(println)
    //    rdd1_union.foreach(println)
    //    rdd1_filter.foreach(println)
    //    rdd1_flatMap.foreach(i => print(i + " "))
    //    rdd2_flatMap.foreach(println)
    //    rdd3_intersection.foreach(i => print(i + " "))
    //    rdd1_subtract.foreach(i => print(i + " "))
    //    rdd1_cartesian.foreach(println)

    //reduce()算子：将RDD传入的数据聚合操作，类型相同，返回的内容是与当前RDD同类型
    //reduce()算子当中不会传递默认值
    //reduce()算子中必须要导入两个参数
    //reduce()算子：先并行处理分区内操作，然后将所获得到的分区内操作的返回值做分区间操作
    val rdd4 = sc.parallelize(Array(1, 2, 3, 4, 5, 6, 7, 8, 9),2)
    //    rdd4.reduce((x,y) => {
    //      println(s"x=$x,y=$y")
    //      x+y
    //    })
    //    println(rdd4_sum)
    //上述计算总和有两个写法
    //    println(rdd1.reduce(_ * _))

    //fold()算子：与reduce()算子相类似，只是多了一个初始值
    //格式：fold(a)(b)
    //a：初始值，b：分区内和分区间操作
    //    println(rdd4.fold(10)(
    //      (x,y) => {
    //        println(s"x=$x,y=$y")
    //        x+y
    //      }
    //    ))

    //aggregate()算子：实现分区内与分区间不同的操作过程，并且返回的RDD类型与原RDD类型可以不相同
    //拥有三个参数，其格式为aggregate(a)(b,c)
    //a：期待返回类型的初始值,b：分区内操作，c：分区间操作
    /*
     * 期待返回类型为键值对，(0,0) => (k,v)，k：总和；v：个数
     * 分区内操作：(x,y) => x是一个二元组，y是指针（可以理解成为reduce()/fold()算子中的y）：x的格式(k,v),k：总和；v：个数
     * 分区间操作：(x,y) => x和y都表示为二元组，实际上x是总和，y是指针
     * * 这个方法格式的缩写：
     * * rdd1.aggregate((0,0))(
     *     (x,y) => x ,
     *     (x,y) => x
     * )
     */
    //    rdd1.aggregate((0,0))(
    //      (x,y) => (x._1 + y , x._2 + 1),
    //      (x,y) => (x._1 + y._1,x._2+y._2)
    //    )
    val rdd1_aggregate = rdd1.aggregate((0,0))(
      (x: (Int,Int),y: Int) => (x._1 + y,x._2 + 1),
      (x: (Int,Int),y: (Int,Int)) => (x._1+y._1,x._2+y._2)
    )
    println("总和为："+rdd1_aggregate._1+"\n个数："+rdd1_aggregate._2+"\n平均值："+(rdd1_aggregate._1/rdd1_aggregate._2))

    //sum()算子：求和算子，将RDD内的数据相加
    //    println(rdd1.sum())

  }
}